package com.swak.algs.paxos;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 议员， 发起提议
 * 
 * @author 618lf
 */
public class Proposer extends IoEvent {

	private static AtomicInteger SN = new AtomicInteger(1);
	private static Logger log = LoggerFactory.getLogger(Proposer.class);
	private ScheduledExecutorService scheduler;

	/**
	 * 当前的编号
	 */
	int b;

	/**
	 * 反馈的最大的编号
	 */
	int maxb;

	/**
	 * 提交的议案
	 */
	Object v;

	/**
	 * ok 的数量， 超过半数则发起下一步
	 */
	int oks;

	/**
	 * 通讯网络
	 */
	NetWork netWork;

	/**
	 * 当前正在进行的步骤
	 */
	boolean isPreparing;
	boolean isAccepting;

	/**
	 * ID : 名称
	 */
	int id;
	String name;

	/**
	 * 初始值
	 */
	Object initValue;

	public Proposer() {
		scheduler = Executors.newSingleThreadScheduledExecutor();
		scheduler.scheduleAtFixedRate(() -> {
			if (!Paxos.me().isFinish()) {
				System.out.println("######################## 开启下一轮    #######################");
				this.propose(this.initValue + "【" + SN.incrementAndGet() + "】");
			}
		}, 10, 10, TimeUnit.SECONDS);
	}

	public Proposer(NetWork netWork) {
		this();
		this.netWork = netWork;
		this.id = SN.getAndIncrement();
		this.name = "提议员" + this.id;
	}

	/**
	 * 提交议案
	 * 
	 * @param proposal
	 */
	public void propose(Object value) {
		synchronized (this) {
			this.initValue = value;
			this.v = value;
			this.maxb = 0;
			this.oks = 0;
			this.b++;
		}
		this.Prepare();
	}

	/**
	 * 准备阶段
	 */
	public void Prepare() {
		synchronized (this) {
			this.isPreparing = true;
			this.isAccepting = false;

			/**
			 * 创建一个议案
			 */
			Proposal proposal = new Proposal();
			proposal.setStep(1); // 标记为Prepare
			proposal.setB(b);
			proposal.setProposerId(this.id);
			proposal.setProposerName(this.name);

			log.debug("提议员 - 准备阶段：{}  编号 {} 初始值 {}", this.name, this.b, this.v);

			/**
			 * 通过网络发布议案
			 */
			this.netWork.sendPrepareMessage(proposal);
		}
	}

	/**
	 * 准备阶段 -- 响应
	 */
	public void PrepareResponse(PrepareResult proposalResult) {
		synchronized (this) {

			// 当前议员是否处理准备阶段
			if (!this.isPreparing) {
				return;
			}

			/*
			 * 记录反馈的值
			 */
			if (proposalResult.getAb() > maxb && proposalResult.getAv() != null) {
				this.maxb = proposalResult.getAb();
				this.v = proposalResult.getAv();
			}

			/*
			 * 记录回复OK的数量
			 */
			if (proposalResult.isOk()) {
				this.oks++;
			}

			/*
			 * 如果是多数派，则发起第二阶段
			 */
			if (Paxos.me().isMajority(this.oks)) {
				log.debug("提议员 - 准备阶段：{} 多数派同意编号 {}， 同意人数 {} ", this.name, this.b, this.oks);
				this.Accept();
			}
		}

	}

	/**
	 * 接受阶段
	 */
	public void Accept() {
		synchronized (this) {
			/*
			 * 阶段性标识
			 */
			this.isPreparing = false;
			this.isAccepting = true;
			this.oks = 0;

			/*
			 * 接受的数据
			 */
			Proposal proposal = new Proposal();
			proposal.setStep(2); // 标记为Prepare
			proposal.setB(b);
			proposal.setV(v);
			proposal.setProposerId(this.id);
			proposal.setProposerName(this.name);

			log.debug("提议员 - 接受阶段：{} 编号 {} - {}", name, b, v);

			/*
			 * 发布接受消息
			 */
			this.netWork.sendAcceptMessage(proposal);
		}

	}

	/**
	 * 接受阶段 -- 响应
	 */
	public void AcceptResponse(AcceptResult proposalResult) {
		synchronized (this) {

			// 当前议员是否处理准备阶段
			if (!this.isAccepting) {
				return;
			}

			/*
			 * 记录回复OK的数量
			 */
			if (proposalResult.isOk()) {
				this.oks++;
			}

			/*
			 * 如果是多数派，结束即可
			 */
			if (Paxos.me().isMajority(this.oks)) {
				this.isPreparing = false;
				this.isAccepting = false;
				this.oks = 0;
				Paxos.me().finish();
				log.debug("提议员 - 完成：{}  编号 {}， 值 {}", this.name, this.b, this.v);
			}
		}
	}
}